package repeater

import (
    "encoding/json"
    "errors"
    "fmt"
    "io/ioutil"
    "os"
    "sync"
    "time"
    "github.com/gin-gonic/gin"
)

const (
    DEFAULT_OFFSET = int64(-2)
)

type RepeaterController struct {
    repeaters  []Repeater
    head       Sender
    offset     map[string]int64
    initialOffset    int64
    interval   int
    commitPath string
    restServer *gin.Engine
    restAddr   string
    wg         sync.WaitGroup
    errors     chan error
    stopping   chan interface{}
    stopped    chan interface{}
    metrics    *Metrics
    Partition  int32
}

func NewRepeaterController(interval int, commitPath, restAddr string) *RepeaterController {
    if commitPath == "" {
        commitPath = "registry.json"
    }
    if interval == 0 {
        interval = 3
    }
    gin.SetMode(gin.ReleaseMode)
    server := gin.Default()
    return &RepeaterController{
        commitPath: commitPath,
        interval:   interval,
        restServer: server,
        stopping:   make(chan interface{}),
        stopped:    make(chan interface{}),
        errors:     make(chan error),
        restAddr:   restAddr,
    }
}

func (r *RepeaterController) loadOffset() (err error) {
    if r.commitPath == "" {
        return
    }
    data := make(map[string]int64)
    content, err := ioutil.ReadFile(r.commitPath)
    if err == nil {
        err = json.Unmarshal(content, &data)
    }
    if os.IsNotExist(err) {
        err = nil
    }
    r.offset = data
    return
}

func (r *RepeaterController) saveOffset() error {
    if r.commitPath == "" {
        return nil
    }
    if len(r.offset) == 0 {
        return fmt.Errorf("no data to commit")
    }
    content, err := json.Marshal(r.offset)
    if err != nil {
        return err
    }
    if err := os.Remove(r.commitPath + ".tmp"); !os.IsNotExist(err) {
        return err
    }
    //write json to filepath.tmp
    if err := ioutil.WriteFile(r.commitPath+".tmp", content, os.FileMode(0644)); err != nil {
        return err
    }
    //rename filepath.tmp to filepath
    if err := os.Rename(r.commitPath+".tmp", r.commitPath); err != nil {
        return err
    }
    return nil
}

func (r *RepeaterController) errorChan() chan<- error {
    return r.errors
}

func (r *RepeaterController) waitGroup() *sync.WaitGroup {
    return &r.wg
}

func (r *RepeaterController) Run() { //presume setUp is called
    if r.restAddr != "" {
        log.Infof("start rest server on %s", r.restAddr)
        go r.restServer.Run(r.restAddr)
    }
    ticker := time.NewTicker(time.Second * time.Duration(r.interval))
    for _, rp := range r.repeaters {
        go rp.Start()
        log.Infof("started %s", rp.Name())
    }
LOOP:
    for {
        select {
        case <-r.stopping:
            log.Info("stop controler, commit first")
            break LOOP
        case <-ticker.C:
            r.Commit()
        case pkg, ok := <-r.head.Output():
            if !ok {
                log.Notice("input should be closed, ignore")
            } else {
                log.Infof("handle data package size [%d]",pkg.count)
               for _, rp := range r.repeaters {
                    err := r.metrics.WrappedHandle(pkg,rp)
                    if err != nil {
                        log.Errorf("handle datapoint error %v, stopping", err)
                        close(r.stopping)
                    }
                }
                //wg := new(sync.WaitGroup)
                //ctrl := make(chan interface{})
                //errors := make(chan error)
                //sw:=time.Now().UnixNano()/(1000*1000)
                //for _, rp := range r.repeaters {
                //    go func() {
                //        wg.Add(1)
                //        defer wg.Done()
                //        if err := rp.Handle(pkg);err != nil {
                //            errors <- fmt.Errorf("%s:%v",rp.Name(),err)
                //        }
                //    }()
                //}
                //go func() {
                //    wg.Wait()
                //    close(ctrl)
                //}()
                //select {
                //case <-ctrl:
                //    log.Infof("dealed with one batch size %d, cost %d (ms)", pkg.count,time.Now().UnixNano()/(1000*1000)-sw)
                //case err := <-errors:
                //    log.Errorf("handle datapoint error %v, stopping", err)
                //    close(r.stopping)
                //}
               //pkg.Reset()
            }
        }
    }
    r.Commit()
    for _, p := range r.repeaters {
        log.Infof("stopping %s", p.Name())
        p.Stop()
        log.Infof("stopped %s", p.Name())
    }
    r.wg.Wait()
    log.Infof("shutdown controler complete...")
    close(r.stopped)
}

func (r *RepeaterController) SetupLinks(repeaters ...Repeater) error {
    if len(repeaters) < 2 {
        return fmt.Errorf("not enough repeaters")
    }
    for i := 0; i < len(repeaters)-1; i++ {
        for j := i + 1; j < len(repeaters); j++ {
            if repeaters[i].Name() == repeaters[j].Name() {
                return errors.New("duplicate repeater")
            }
        }
    }
    for _, rp := range repeaters {
        rp.SetController(r)
        r.repeaters = append(r.repeaters, rp)
    }
    if head, ok := r.repeaters[0].(Sender); !ok {
        return fmt.Errorf("no valid sender")
    } else {
        r.head = head
    }
    return nil
}


func (r *RepeaterController) Stopchan() <-chan interface{} {
   return r.stopped
}

func (r *RepeaterController) Stopped() bool {
    select {
    case <-r.stopped:
        return true
    default:
        return false
    }
}

func (r *RepeaterController) Stop() {
    close(r.stopping)
    <-r.stopped
}

func (r *RepeaterController) Commit() {
    log.Info("start commit...")
    offsets := make(map[string]int64)
    for _, p := range r.repeaters {
        log.Infof("start commit %s", p.Name())
        o, err := r.metrics.WrappedCommit(p)
        if err != nil {
            log.Fatalf("commit %s error: %v", p.Name(), err)
            //r.errors <- err
            //return
            os.Exit(3)
        }
        log.Infof("commit %s result is %d", p.Name(), o)
        offsets[p.Name()] = o
    }
    r.offset = offsets
    if err := r.saveOffset(); err != nil {
        log.Errorf("save offset error %v", err)
        r.errors <- err
        return
    }
    log.Info("complete commit")
}

func (r *RepeaterController) Init() (err error) {
    if err := r.loadOffset(); err != nil {
        return err
    }
    o := DEFAULT_OFFSET
    for _, v := range r.offset {
        if o < 0 || o > v {
            o = v
        }
    }
    log.Infof("loaded: %v", r.offset)
    log.Infof("satrting  from offset %d", o)
    r.initialOffset=o
    //r.head.SetOffset(o)
    for _, p := range r.repeaters {
        if err = p.Init(); err != nil {
            err=fmt.Errorf("%s %v",p.Name(),err)
            return
        }
    }
    r.metrics = NewMetrics(r)
    log.Infof("init complete")
    return nil
}
